package com.ibm.cps.processors;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.join.AbstractJoiner;
import com.ibm.cps.join.JoinerFactory;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessageBuilder;
import com.ibm.cps.newmessage.BasicDataSourceMetadata;
import com.ibm.cps.processors.vector.SchemaUtility;
import com.ibm.cps.sql.expression.ExpressionNode;
import com.ibm.cps.sql.expression.FieldEvaluator;
import com.ibm.cps.sql.expression.ValueExpression;
import com.ibm.cps.sql.expression.function.meta.IgnoredInFunctionCall;
import com.ibm.cps.sql.expression.function.meta.ValueParam;
import com.ibm.datamodel.timeseries.TimeSeriesSchema;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.DataTypes;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

@Info(Name=MessageFields.STATIC_JOINT,Type=Info.ProcessorType.Other)
public class StaticJointProcessor extends SchemaInheritiveProcessor{

	
	@ValueParam
	public String metadataString;
	
	private AbstractJoiner joiner;
	private ITimeSeriesSchema staticschema;	
	
	public StaticJointProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema)
			throws CPSException {
		super(tenantId, processorId, source, parentSchema);
	
	}


	@Override
	protected List<NewField> defineFieldsToBeAdded() {
		List<NewField> fields = new ArrayList<>();
		for (int i = 0; i < staticschema.getTagCount(); i++) {
			String tag = staticschema.getTagNames()[i];
			byte datatype = staticschema.getDataTypes()[i];
			NewField field = new NewField(tag, DataTypes.getDataType(datatype));
			fields.add(field);
		}
		
		return fields;
	}


	@Override
	public void dispose() {
		try {
			joiner.close();
		} catch (CPSException e) {
			e.printStackTrace();
		}
	}

	@Override
	protected Collection<ParsedDataMessage> runProcessorLogic(ITimeSeriesSchema parentSchema,
			ITimeSeriesSchema outputSchema, ParsedDataMessage message) throws CPSException {
		
		Collection<ParsedDataMessage> messages = new ArrayList<>();
		ParsedDataMessageBuilder outputBuilder = ParsedDataMessageBuilder.createInhertiveBuilder(outputSchema, message, parentSchema);
		
		Comparable key = message.getTsKey();
		ObjectNode query = JSONUtility.newObjectNode();
		if (key instanceof Long) {
			query.put(MessageFields.TSKEY, (Long)key);
		}else if (key instanceof String) {
			query.put(MessageFields.TSKEY, (String)key);
		}else {
			throw new CPSException(ErrorCode.ERROR_FIELD_TYPE, "unknown type for tskey " + key.getClass().getName());
		}

		Collection<ParsedDataMessage> hit_messages = joiner.query(query);
		
		// merge the hit_message and input Message
		for(ParsedDataMessage hit : hit_messages){
			String[] tagNames = staticschema.getTagNames();
			for (int i = 0; i < tagNames.length; i++) {
				String tagName = tagNames[i];
				byte typeByte = staticschema.getDataType(i);
				Object value;
				if(typeByte == DataTypes.DT_DOUBLE_ARRAY || typeByte == DataTypes.DT_INT_ARRAY || typeByte == DataTypes.DT_STRING_ARRAY)
					value = hit.getArrayValue(i);
				else
					value = hit.getValue(i);
				outputBuilder.append(tagName, value);
			}
			
			messages.add(outputBuilder.toParsedDataMessage());
		}
		
		return messages;
	}


	@Override
	public void initialize() throws CPSException {
		joiner = JoinerFactory.createJoiner(metadataString);
		staticschema = joiner.getSchema();
	}
	
	public ITimeSeriesSchema getStaticSchema(){
		return staticschema;
	}

}
